package com.example.feature.demo.controller;

import com.example.feature.demo.dto.Address;
import com.example.feature.demo.dto.FileDto;
import com.example.feature.demo.dto.Foo1;
import com.example.feature.demo.dto.Person;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;

/**
 * KafkaTestController
 *
 * @author zhouxianfeng
 * @date 2021-5-20
 */
@Slf4j
@RestController()
@RequestMapping("/kafka")
public class KafkaTestController {

    @Autowired
    private KafkaTemplate<Object, Object> template;


    @PostMapping("/person")
    public void sendPerson(@RequestBody Person person) {
        template.send("topic-zhou1", person);
    }

    @PostMapping("/address")
    public void sendAddress(@RequestBody Address address) {
        template.send("topic-zhou2", address);
    }

    @PostMapping("/object")
    public void sendObject(@RequestBody FileDto dto) {
        template.send("topic-zhou2", dto);
    }

    @PostMapping(path = "/foos/{what}")
    public void sendFoos(@PathVariable String what) {
        this.template.executeInTransaction(kafkaTemplate -> {
            StringUtils.commaDelimitedListToSet(what).stream()
                    .map(s -> new Foo1(s))
                    .forEach(foo -> kafkaTemplate.send("topic-zhou3", foo));
            return null;
        });
    }

    @KafkaListener(id = "fooGroup323", topics = "topic-zhou3")
    public void listen1(Foo1 foo) throws IOException, InterruptedException {

        log.info("Received: {}", foo);
    }


}
